/*

This file is part of roomyak.

Copyright (C) 2008 Pixecur Technologies Incorporated (Esotera).
Visit Esotera at http://www.esoteras3.com
 
roomyak is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 3, or (at your option)
any later version.
 
roomyak is distributed in the hope that it will be useful, but
WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
General Public License for more details.
 
You should have received a copy of the GNU General Public License
along with roomyak; see the file COPYING.  If not, write to 
Pixecur Technologies, 213-31 Peet Street, St. John's, NL, A1B 3W8. 
 
Linking this library statically or dynamically with other modules is
making a combined work based on this library.  Thus, the terms and
conditions of the GNU General Public License cover the whole
combination.
 
*/

package com.datasentinel.util.concurrent;

import gnu.trove.TLongArrayList;
import gnu.trove.TLongObjectHashMap;
import gnu.trove.TLongObjectIterator;
import gnu.trove.TLongObjectProcedure;
import gnu.trove.TLongProcedure;

import java.util.LinkedList;
import java.util.Queue;

import com.datasentinel.util.cache.UnboundedItemCache;


public class ReadWriteLockMap
{
	private static final long LOCK_EXPIRATION_TIMEOUT = 4 * 60 * 1000;// 4 minutes

	private static final int LOCK_CACHE_SIZE = 1024;

	static class LockDescriptorData implements MutableLockDescriptor, Comparable<LockDescriptorData>
	{
		protected long lockId;
		protected long unlockKey;
		protected long lockTime;

		protected LockMode mode;
		protected long lockOwner;

		public LockDescriptorData()
		{
		}

		public void setLockOwner(long lockOwner)
		{
			this.lockOwner = lockOwner;
		}

		public void setMode(LockMode mode)
		{
			this.mode = mode;
		}

		@Override
		public long getOwnerId()
		{
			return lockOwner;
		}

		public LockMode getMode()
		{
			return mode;
		}

		@Override
		public String toString()
		{
			return "Lock [" + getMode() + " by " + getOwnerId() + "]";
		}

		public void reset()
		{
			this.mode = null;
			this.lockId = 0l;
			this.lockOwner = 0l;
			this.unlockKey = 0l;
		}

		@Override
		public int compareTo(LockDescriptorData o)
		{
			return this.hashCode() - o.hashCode();
		}
	}

	public static MutableLockDescriptor createLockInstance()
	{
		return new LockDescriptorData();
	}


	private static class LockState
	{
		protected TLongObjectHashMap<LockDescriptorData> ownerLockMap;
		protected TLongObjectHashMap<LockDescriptorData> ownerPendingWriteMap;
		protected LinkedList<LockDescriptorData> pendingWriteQueue;
	}

	private UnboundedItemCache<LockDescriptorData> lockCache;

	private TLongObjectHashMap<LockState> lockMap;


	public ReadWriteLockMap()
	{


		lockMap = new TLongObjectHashMap<LockState>();

		lockCache = new UnboundedItemCache<LockDescriptorData>(LOCK_CACHE_SIZE) {
			@Override
			protected LockDescriptorData createItem()
			{
				return new LockDescriptorData();
			}

			@Override
			protected void resetItem(LockDescriptorData item)
			{
				item.reset();
			}
		};
	}

	/**
	 * @return the lockOwner.longValue on success or ownerId on failure
	 */
	public synchronized LockDescriptor tryReadLock(long lockId, long lockOwner)
	{
		LockState lockState = lockMap.get(lockId);
		TLongObjectHashMap<LockDescriptorData> ownerLockMap = lockState == null ? null : lockState.ownerLockMap;
		LinkedList<LockDescriptorData> pendingWriteQueue = lockState == null ? null : lockState.pendingWriteQueue;

		cleanupExpiredLocks(lockId, lockOwner);

		if (pendingWriteQueue == null || pendingWriteQueue.isEmpty())
		{
			if (ownerLockMap == null || ownerLockMap.isEmpty())
			{
				LockDescriptor resultLock = addNewLock(LockMode.READ, lockId, lockOwner, 0);
				return resultLock;
			}

			//The same owner can relock multiple times and change the mode of the lock between READ and WRITE
			LockDescriptorData lock = ownerLockMap.get(lockOwner);
			if (lock != null)
			{
				/*
				 * If the file was locked in WRITE mode, it should remain so. No need to downgrade it. 
				 */
				setupLock(lock, lock.mode, lockId, lockOwner, lock.unlockKey);
				return lock;
			}

			TLongObjectIterator<LockDescriptorData> lockIterator = ownerLockMap.iterator();
			lockIterator.advance();
			LockDescriptor firstLockDescriptor = lockIterator.value();
			if (firstLockDescriptor.getMode().equals(LockMode.READ))
			{
				LockDescriptor resultLock = addNewLock(LockMode.READ, lockId, lockOwner, 0);
				return resultLock;
			}

			return firstLockDescriptor;
		}
		else
		{
			LockDescriptorData lock = lockState.ownerPendingWriteMap.get(lockOwner);
			if (lock != null)
			{
				return lock;
			}
		}

		return pendingWriteQueue.get(0);
	}

	/**
	 * @return the lockOwner.longValue on success or ownerId on failure. If the lock is
	 * READ locked, returns the first lock owner.
	 */
	public synchronized LockDescriptor tryWriteLock(long lockId, final long lockOwner, long unlockKey)
	{
		LockDescriptorData lock = null;
		LockState lockState = lockMap.get(lockId);
		TLongObjectHashMap<LockDescriptorData> ownerLockMap = lockState == null ? null : lockState.ownerLockMap;
		Queue<LockDescriptorData> pendingWriteQueue = lockState == null ? null : lockState.pendingWriteQueue;

		cleanupExpiredLocks(lockId, lockOwner);

		if ((pendingWriteQueue == null) || pendingWriteQueue.isEmpty())
		{
			if (ownerLockMap == null || ownerLockMap.isEmpty())
			{
				LockDescriptor resultLock = addNewLock(LockMode.WRITE, lockId, lockOwner, unlockKey);
				return resultLock;
			}

			//The same owner can relock multiple times and change the mode of the lock between READ and WRITE.
			//The user can change the mode of the lock to WRITE only if she is the only one who has locked the lockId.
			lock = ownerLockMap.get(lockOwner);
			if (ownerLockMap.size() == 1 && lock != null)
			{

				if (lock.mode == LockMode.READ || lock.unlockKey == unlockKey)
				{
					/*
					 * The owner can upgrade the lock to WRITE if there are no other readers or pending writes.
					 */
					setupLock(lock, LockMode.WRITE, lockId, lockOwner, unlockKey);
					return lock;
				}
				else
				{
					/*
					 * The owner cannot lock twice with different unlockKey-s.
					 */

					return new LockDescriptor(){
						private final long owner = lockOwner;
						@Override
						public LockMode getMode()
						{
							return LockMode.WRITE_INVALID_KEY;
						}
						
						@Override
						public long getOwnerId()
						{
							return owner;
						}
					};
				}
			}
		}
		else
		{
			/*
			 * If there are pending locks in the queue, it does not matter whether ownerLockMap is 
			 * empty or not - only the first owner in the queue can obtain the lock, and only 
			 */
			LockDescriptor pendingWriteLock = pendingWriteQueue.peek();
			if (pendingWriteLock.getOwnerId() == lockOwner && ownerLockMap.size() == 0)
			{
				if (ownerLockMap == null || ownerLockMap.isEmpty())
				{
					pullPendingWriteLock(lockState);

					LockDescriptor resultLock = addNewLock(LockMode.WRITE, lockId, lockOwner, unlockKey);
					return resultLock;
				}

				/*
				 * 1. The same owner will never be represented twice in the pendingWriteQueue.
				 */

				/*
				 * It is possible that:
				 * 		a) owner locked the file in READ mode 
				 * 		b) other users locked in READ mode
				 * 		c) owner added a pending write lock
				 * 		d) other users released their READ locks
				 * 		e) there is a single READ lock on the file and pending lock in the queue
				 */
				lock = ownerLockMap.get(lockOwner);
				if (ownerLockMap.size() == 1 && lock != null)
				{
					pullPendingWriteLock(lockState);

					/*
					 * The owner can upgrade the lock to WRITE if there are no other readers or pending writes.
					 */
					setupLock(lock, LockMode.WRITE, lockId, lockOwner, unlockKey);
					return lock;
				}
			}
		}

		LockDescriptor resultLock = refreshPendingWriteLock(lockState, lockId, lockOwner, unlockKey);
		return resultLock;
	}


	public synchronized boolean refreshReadLock(long lockId, long lockOwner)
	{
		LockState lockContainer = lockMap.get(lockId);
		TLongObjectHashMap<LockDescriptorData> ownerLockMap = lockContainer == null ? null : lockContainer.ownerLockMap;
		if (ownerLockMap == null)
			return false;
		LockDescriptorData lock = ownerLockMap.get(lockOwner);

		if (lock != null && tryReadLock(lock.lockId, lock.lockOwner).getOwnerId() != lock.lockOwner)
		{
			return false;
		}
		else if (lock == null)
		{
			return false;
		}

		return true;
	}

	public synchronized boolean refreshWriteLock(long lockId, long lockOwner)
	{
		LockState lockContainer = lockMap.get(lockId);
		TLongObjectHashMap<LockDescriptorData> ownerLockMap = lockContainer == null ? null : lockContainer.ownerLockMap;

		if (ownerLockMap == null)
			return false;
		LockDescriptorData lock = ownerLockMap.get(lockOwner);

		if (lock != null && tryWriteLock(lock.lockId, lock.lockOwner, lock.unlockKey).getOwnerId() != lock.lockOwner)
		{
			return false;
		}
		else if (lock == null)
		{
			return false;
		}

		return true;
	}

	/**
	 * Return true if the lock was successfully defeated, expired or did not exist at all.
	 * If the lock had different owner or the unlockKey is wrong the lock is not removed and
	 * returns false;
	 */
	public synchronized boolean unlock(long lockId, long lockOwner, long unlockKey)
	{
		LockState lockContainer = lockMap.get(lockId);
		TLongObjectHashMap<LockDescriptorData> ownerLockMap = lockContainer == null ? null : lockContainer.ownerLockMap;
		if (ownerLockMap == null)
			return true;

		if (ownerLockMap.isEmpty())
		{
			lockMap.remove(lockId);
			return true;
		}

		LockDescriptorData lock = ownerLockMap.get(lockOwner);
		if (lock == null)
			return false;

		if (lock.mode.equals(LockMode.WRITE))
		{
			if (lock.unlockKey != unlockKey)
			{
				if (!isLockExpired(lock))
				{
					return false;
				}
			}
		}
		removeLock(lockId, lockOwner);

		return true;
	}

	public synchronized boolean removeLock(long lockId, long lockOwner)
	{
		LockState lockContainer = lockMap.get(lockId);
		TLongObjectHashMap<LockDescriptorData> ownerLockMap = lockContainer == null ? null : lockContainer.ownerLockMap;
		LinkedList<LockDescriptorData> pendingWriteQueue = lockContainer == null ? null
				: lockContainer.pendingWriteQueue;
		if (ownerLockMap == null)
			return true;

		LockDescriptorData lock = ownerLockMap.get(lockOwner);
		if (lock == null)
			return false;

		ownerLockMap.remove(lockOwner);

		if (ownerLockMap.isEmpty() && (pendingWriteQueue == null || pendingWriteQueue.isEmpty()))
		{
			lockMap.remove(lockId);
		}

		lockCache.returnItem(lock);

		return true;
	}

	private void cleanupExpiredLocks(long lockId, long ownerId)
	{
		LockState lockState = lockMap.get(lockId);
		if (lockState == null)
		{
			return;
		}

		final TLongObjectHashMap<LockDescriptorData> ownerLockMap = lockState.ownerLockMap;
		if (ownerLockMap != null && !ownerLockMap.isEmpty())
		{
			final TLongArrayList expiredLockOwnerList = new TLongArrayList();

			ownerLockMap.forEachEntry(new TLongObjectProcedure<LockDescriptorData>() {
				@Override
				public boolean execute(long key, LockDescriptorData value)
				{
					if (isLockExpired(value))
					{
						expiredLockOwnerList.add(key);
					}

					return true;
				}
			});

			expiredLockOwnerList.forEach(new TLongProcedure() {
				@Override
				public boolean execute(long expiredOwnerId)
				{
					ownerLockMap.remove(expiredOwnerId);
					return true;
				}
			});
		}

		LinkedList<LockDescriptorData> pendingWriteQueue = lockState.pendingWriteQueue;
		if (pendingWriteQueue != null && !pendingWriteQueue.isEmpty())
		{
			int queueSize = pendingWriteQueue.size();
			LockDescriptorData pendingWriteLock = null;
			for (int i = 0; i < queueSize; ++i)
			{
				pendingWriteLock = pendingWriteQueue.get(i);
				if (isLockExpired(pendingWriteLock))
				{
					pendingWriteQueue.remove(i);
					lockState.ownerPendingWriteMap.remove(pendingWriteLock.lockOwner);
					queueSize--;
					continue;
				}
			}
		}

	}

	private void setupLock(LockDescriptorData lock, LockMode mode, long lockId, long lockOwner, long unlockKey)
	{
		lock.lockId = lockId;
		lock.mode = mode;
		lock.lockTime = System.currentTimeMillis();
		lock.lockOwner = lockOwner;
		lock.unlockKey = unlockKey;
	}

	private void pullPendingWriteLock(LockState lockState)
	{
		LockDescriptorData pendingWriteLock = lockState.pendingWriteQueue.poll();
		lockState.ownerPendingWriteMap.remove(pendingWriteLock.lockOwner);
		lockCache.returnItem(pendingWriteLock);
	}

	private LockDescriptor refreshPendingWriteLock(LockState lockState, long lockId, long lockOwner, long unlockKey)
	{
		LinkedList<LockDescriptorData> pendingWriteQueue = lockState.pendingWriteQueue;
		TLongObjectHashMap<LockDescriptorData> ownerPendingWriteMap = lockState.ownerPendingWriteMap;

		LockDescriptorData pendingWriteLock = null;
		if (pendingWriteQueue == null)
		{
			pendingWriteQueue = lockState.pendingWriteQueue = new LinkedList<LockDescriptorData>();
		}

		if (ownerPendingWriteMap == null)
		{
			ownerPendingWriteMap = lockState.ownerPendingWriteMap = new TLongObjectHashMap<LockDescriptorData>();
		}
		else
		{
			pendingWriteLock = ownerPendingWriteMap.get(lockOwner);
		}

		if (pendingWriteLock == null)
		{
			pendingWriteLock = lockCache.getItem();
			ownerPendingWriteMap.put(lockOwner, pendingWriteLock);
			pendingWriteQueue.add(pendingWriteLock);
		}

		setupLock(pendingWriteLock, LockMode.PENDING_WRITE, lockId, lockOwner, unlockKey);

		return pendingWriteLock;
	}

	private LockDescriptor addNewLock(LockMode mode, long lockId, long lockOwner, long unlockKey)
	{
		LockDescriptorData lock = null;
		LockState lockState = lockMap.get(lockId);
		if (lockState == null)
		{
			lockState = new LockState();
			lockState.ownerLockMap = new TLongObjectHashMap<LockDescriptorData>();

			lockMap.put(lockId, lockState);
		}

		TLongObjectHashMap<LockDescriptorData> ownerLockMap = lockState.ownerLockMap;

		lock = lockCache.getItem();

		setupLock(lock, mode, lockId, lockOwner, unlockKey);

		ownerLockMap.put(lockOwner, lock);

		return lock;
	}

	private boolean isLockExpired(LockDescriptorData lock)
	{
		return (System.currentTimeMillis() - lock.lockTime) >= LOCK_EXPIRATION_TIMEOUT;
	}

}
